package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

public abstract class DimMapFunction<T> extends RichMapFunction<T, T> {

    private Jedis jedis;
    private Connection hBaseConnection;

    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConnection = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hBaseConnection);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public T map(T t) throws Exception {
        // 根据流中对象获取要关联的维度主键
        String key = getRowKey(t);
        // 先从redis缓存中好去维度数据
        JSONObject dimJsonObj = RedisUtil.readDim(jedis, getTableName(), key);
        if (dimJsonObj != null) {
            // 如果获取到，直接将其作为返回值返回（缓存命中）
            System.out.println("从Redis中获得" + getTableName() + "表的" + key+ "维度数据");
        } else {
            // 如果没有获取到，发送请求到HBase中查询维度
            dimJsonObj = HBaseUtil.getRow(hBaseConnection, Constant.HBASE_NAMESPACE, getTableName(), key, JSONObject.class);
            if (dimJsonObj != null) {
                System.out.println("从HBase中获得" + getTableName() + "表的" + key+ "维度数据");
                // 将查询结果放到Redis
                RedisUtil.writeDim(jedis, getTableName(), key, dimJsonObj);
            } else {
                System.out.println("HBase中没有" + getTableName() + "表的维度数据");
            }
        }
        // 将查询出来的维度数据补充到流对象上
        if (dimJsonObj != null) {
            addDims(t, dimJsonObj);
        }
        return null;

    }

    public abstract void addDims(T t, JSONObject dimJsonObj);
    public abstract String getRowKey(T t);
    public abstract String getTableName();
}
